package com.song.storm_demo.hbase_chap8.consumer;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.util.Bytes;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

public class SaveBolt implements IBasicBolt{
	
	
	private Configuration conf ;
	private HTable htable;
	
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("hbase","ncount"));
		
	}

	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}

	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	public void execute(Tuple tuple, BasicOutputCollector collector) {
		
		String line = tuple.getString(0);
		int   count  = tuple.getInteger(1);
		
		try {
			htable.incrementColumnValue(Bytes.toBytes(line), Bytes.toBytes("col"),Bytes.toBytes("data"), count);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	public void prepare(Map arg0, TopologyContext arg1) {
		
		conf = HBaseConfiguration.create();
		try {
			htable = new HTable(conf, "storm_hbase_kafka");
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

}
